/*
 * MIT License
 *
 * Copyright (c) 2023 北京凯特伟业科技有限公司
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */
package com.je.bpm.engine.impl.asyncexecutor;

import com.je.bpm.engine.ActivitiOptimisticLockingException;
import com.je.bpm.engine.delegate.event.ActivitiEventType;
import com.je.bpm.engine.delegate.event.impl.ActivitiEventBuilder;
import com.je.bpm.engine.impl.cfg.ProcessEngineConfigurationImpl;
import com.je.bpm.engine.impl.cmd.ExecuteAsyncJobCmd;
import com.je.bpm.engine.impl.cmd.LockExclusiveJobCmd;
import com.je.bpm.engine.impl.cmd.UnlockExclusiveJobCmd;
import com.je.bpm.engine.impl.context.Context;
import com.je.bpm.engine.impl.interceptor.Command;
import com.je.bpm.engine.impl.interceptor.CommandConfig;
import com.je.bpm.engine.impl.interceptor.CommandContext;
import com.je.bpm.engine.impl.jobexecutor.FailedJobCommandFactory;
import com.je.bpm.engine.impl.persistence.entity.JobEntity;
import com.je.bpm.engine.runtime.Job;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 */
public class ExecuteAsyncRunnable implements Runnable {

    private static Logger log = LoggerFactory.getLogger(ExecuteAsyncRunnable.class);

    protected String jobId;
    protected Job job;
    protected ProcessEngineConfigurationImpl processEngineConfiguration;

    public ExecuteAsyncRunnable(String jobId, ProcessEngineConfigurationImpl processEngineConfiguration) {
        this.jobId = jobId;
        this.processEngineConfiguration = processEngineConfiguration;
    }

    public ExecuteAsyncRunnable(Job job, ProcessEngineConfigurationImpl processEngineConfiguration) {
        this.job = job;
        this.jobId = job.getId();
        this.processEngineConfiguration = processEngineConfiguration;
    }

    @Override
    public void run() {
        if (job == null) {
            job = processEngineConfiguration.getCommandExecutor().execute(new Command<JobEntity>() {
                @Override
                public JobEntity execute(CommandContext commandContext) {
                    return commandContext.getJobEntityManager().findById(jobId);
                }
            });
        }

        boolean lockNotNeededOrSuccess = lockJobIfNeeded();

        if (lockNotNeededOrSuccess) {
            executeJob();
            unlockJobIfNeeded();
        }

    }

    protected void executeJob() {
        try {
            processEngineConfiguration.getCommandExecutor().execute(new ExecuteAsyncJobCmd(jobId));
        } catch (final ActivitiOptimisticLockingException e) {

            handleFailedJob(e);

            if (log.isDebugEnabled()) {
                log.debug("Optimistic locking exception during job execution. If you have multiple async executors running against the same database, "
                        + "this exception means that this thread tried to acquire an exclusive job, which already was changed by another async executor thread."
                        + "This is expected behavior in a clustered environment. " + "You can ignore this message if you indeed have multiple job executor threads running against the same database. "
                        + "Exception message: {}", e.getMessage());
            }

        } catch (Throwable exception) {
            handleFailedJob(exception);

            // Finally, Throw the exception to indicate the ExecuteAsyncJobCmd failed
            String message = "Job " + jobId + " failed";
            log.error(message, exception);
        }
    }

    protected void unlockJobIfNeeded() {
        try {
            if (job.isExclusive()) {
                processEngineConfiguration.getCommandExecutor().execute(new UnlockExclusiveJobCmd(job));
            }

        } catch (ActivitiOptimisticLockingException optimisticLockingException) {
            if (log.isDebugEnabled()) {
                log.debug("Optimistic locking exception while unlocking the job. If you have multiple async executors running against the same database, "
                        + "this exception means that this thread tried to acquire an exclusive job, which already was changed by another async executor thread."
                        + "This is expected behavior in a clustered environment. " + "You can ignore this message if you indeed have multiple job executor acquisition threads running against the same database. "
                        + "Exception message: {}", optimisticLockingException.getMessage());
            }

        } catch (Throwable t) {
            log.error("Error while unlocking exclusive job " + job.getId(), t);
        }
    }

    /**
     * Returns true if lock succeeded, or no lock was needed.
     * Returns false if locking was unsuccessfull.
     */
    protected boolean lockJobIfNeeded() {
        try {
            if (job.isExclusive()) {
                processEngineConfiguration.getCommandExecutor().execute(new LockExclusiveJobCmd(job));
            }

        } catch (Throwable lockException) {
            if (log.isDebugEnabled()) {
                log.debug("Could not lock exclusive job. Unlocking job so it can be acquired again. Catched exception: " + lockException.getMessage());
            }

            // Release the job again so it can be acquired later or by another node
            unacquireJob();

            return false;
        }

        return true;
    }

    protected void unacquireJob() {
        CommandContext commandContext = Context.getCommandContext();
        if (commandContext != null) {
            commandContext.getJobManager().unacquire(job);
        } else {
            processEngineConfiguration.getCommandExecutor().execute(new Command<Void>() {
                @Override
                public Void execute(CommandContext commandContext) {
                    commandContext.getJobManager().unacquire(job);
                    return null;
                }
            });
        }
    }

    protected void handleFailedJob(final Throwable exception) {
        processEngineConfiguration.getCommandExecutor().execute(new Command<Void>() {

            @Override
            public Void execute(CommandContext commandContext) {
                CommandConfig commandConfig = processEngineConfiguration.getCommandExecutor().getDefaultConfig().transactionRequiresNew();
                FailedJobCommandFactory failedJobCommandFactory = commandContext.getFailedJobCommandFactory();
                Command<Object> cmd = failedJobCommandFactory.getCommand(job.getId(), exception);

                log.trace("Using FailedJobCommandFactory '" + failedJobCommandFactory.getClass() + "' and command of type '" + cmd.getClass() + "'");
                processEngineConfiguration.getCommandExecutor().execute(commandConfig, cmd);

                // Dispatch an event, indicating job execution failed in a
                // try-catch block, to prevent the original exception to be swallowed
                if (commandContext.getEventDispatcher().isEnabled()) {
                    try {
                        commandContext.getEventDispatcher().dispatchEvent(ActivitiEventBuilder.createEntityExceptionEvent(ActivitiEventType.JOB_EXECUTION_FAILURE, job, exception));
                    } catch (Throwable ignore) {
                        log.warn("Exception occurred while dispatching job failure event, ignoring.", ignore);
                    }
                }

                return null;
            }

        });
    }

}
